package com.loyalove.demo.akka

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._

class SparkWorker(masterUrl: String) extends Actor {

  val workerId = UUID.randomUUID.toString

  // master Ref
  var masterProxy: ActorSelection = _

  override def preStart(): Unit = {
    masterProxy = context.actorSelection(masterUrl)
  }

  override def receive: Receive = {

    // 自己已就绪
    case "started" => {
      // worker向master注册自己的信息
      masterProxy ! RegisterWorkerInfo(workerId, 2, 8 * 1024)
    }
    // master发送的注册成功消息
    case RegisteredWorkerInfo(workId, core, ram) => {
      // worker 启动定时器,定时向master发送心跳
      import context.dispatcher
      context.system.scheduler
        .schedule(0 millis, 1500 millis, self, SendHeartbeat)

    }
    // 发送心跳消息
    case SendHeartbeat => {
      println(s"------ $workerId start send heartbeat ------")
      masterProxy ! Heartbeat(workerId)
    }
  }
}

object SparkWorker {

  def main(args: Array[String]): Unit = {

    if (args.length != 4) {
      println(
        s"""
           |请输入参数： <host> <port> <workName> <masterUrl>
        """.stripMargin)
      sys.exit()
    }

    val host = args(0)
    val port = args(1)
    val workName = args(2)
    val masterUrl = args(3)

    val config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider=akka.remote.RemoteActorRefProvider
         |akka.remote.netty.tcp.hostname=$host
         |akka.remote.netty.tcp.port=$port
         """.stripMargin)

    val actorSystem = ActorSystem("sparkWorker", config)

    // 创建自己的actorRef对象
    val workerActorRef = actorSystem.actorOf(Props(new SparkWorker(masterUrl)), workName)

    // 就绪消息发送
    workerActorRef ! "started"
  }

}
